package com.kingpeng.rxjava_sample;

import android.content.Intent;
import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;
import android.widget.Button;
import android.widget.LinearLayout;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.Observable;
import java.util.Observer;
import java.util.concurrent.TimeUnit;

import butterknife.BindView;
import butterknife.ButterKnife;
import butterknife.OnClick;
import io.reactivex.Flowable;
import io.reactivex.FlowableSubscriber;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiConsumer;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.Subject;

public class RxJavaAty extends AppCompatActivity {

    private static final String TAG = "#RxJavaAty";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_rxjava);
        ButterKnife.bind(this);
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        if (mDisposable != null) {
            mDisposable.dispose();
        }
    }

    @OnClick(R.id.btn_from)
    public void onClickFrom() {
        //just( ) 和from( ) 方法将对象，列表，对象属性转换为发射那些对象的Observable
        // 转换后的Observable每发射一项数据，会同步地调用任何订阅者的onNext()方法，最后会调用订阅者的onCompleted()方法
        io.reactivex.Observable.fromArray("a", "b", "c").observeOn(Schedulers.io()).doOnNext(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                logD("doOnNext onNext : " + s);
            }
        }).subscribe(new io.reactivex.Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                logD("onSubscribe : " + d.toString());
            }

            @Override
            public void onNext(String s) {
                logD("onNext : " + s);
            }

            @Override
            public void onError(Throwable e) {
                logD("onError : " + e.toString());
            }

            @Override
            public void onComplete() {
                logD("onComplete : ");
            }
        });
    }

    Disposable mDisposable;

    @OnClick(R.id.btn_interval)
    public void onClickInterval() {
        //定时任务
        if (mDisposable != null) {
            mDisposable.dispose();
        }
        mDisposable = Flowable.interval(1, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        logD("doOnNext value : " + aLong);
                    }
                }).subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        logD("get value : " + aLong);
                    }
                });
    }

    @OnClick(R.id.btn_zip)
    public void onClickZip() {
        // zip 操作符可以将多个 Observable 的数据结合为一个数据源再发射出去
        io.reactivex.Observable<Integer> o1 = io.reactivex.Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                logD("------------o1:subscribe----------");
                e.onNext(1);
                e.onNext(2);
            }
        });

        io.reactivex.Observable<Integer> o2 = io.reactivex.Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                logD("------------o2:subscribe----------");
                e.onNext(21);
                e.onNext(22);
            }
        });
        io.reactivex.Observable.zip(o1, o2, new BiFunction<Integer, Integer, String>() {
            @Override
            public String apply(Integer integer, Integer integer2) throws Exception {
                return "o" + integer + " * " + "o" + integer2;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                logD("Consuer-accept : " + s);
            }
        });
    }

    @OnClick(R.id.btn_concat)
    public void onClickConcat() {
        //concat 的必须调用 onComplete 后才能订阅下一个 Observable, 按顺序发射内容给观察者
        ObservableOnSubscribe<String> o1 = new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                Log.d(TAG, "Observable-subscribe-o1: threadName: " + Thread.currentThread().getName());
                e.onNext("o1-1");
                e.onNext("o1-2");
                logD("Observable-subscribe-o1: complete");
                e.onComplete();
                logD("Observable-subscribe-o1: o1-3");
                e.onNext("o1-3");
            }
        };

        ObservableOnSubscribe<String> o2 = new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                Log.d(TAG, "Observable-subscribe-o2: threadName: " + Thread.currentThread().getName());
                e.onNext("o2-1");
                e.onNext("o2-2");
                e.onNext("o2-3");
            }
        };

        io.reactivex.Observable.concat(io.reactivex.Observable.create(o1), io.reactivex.Observable.create(o2)).subscribeOn(Schedulers.newThread()).observeOn(Schedulers.io())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.d(TAG, "Consumer-accept: threadName: " + Thread.currentThread().getName());
                        logD(s);
                    }
                });
    }

    @OnClick(R.id.btn_map)
    public void onClickMap() {
        //map 操作符可以将一个 Observable 对象通过某种关系转换为另一个Observable 对象
        io.reactivex.Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                Log.d(TAG, "subscribe: 1000 threadName: " + Thread.currentThread().getName());
                e.onNext(10000);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                Log.d(TAG, "apply-Function: threadName: " + Thread.currentThread().getName());
                return "map change Integer to String , the value is " + integer;
            }
        }).subscribeOn(Schedulers.io()).observeOn(Schedulers.newThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.d(TAG, "accept-String: threadName: " + Thread.currentThread().getName());
                        logD("consuer-String: accept: " + s);
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.d(TAG, "accept-throwable: threadName: " + Thread.currentThread().getName());
                        logD("consuer-throwable: accept-throwable: " + throwable.toString());
                    }
                });
    }

    @OnClick(R.id.btn_observable)
    public void onClickObservable() {
        // Observable.subcribe(Observer) 生命周期
        // onSubscribe->subscribe->onNext->onComplete/onError
        io.reactivex.Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                logD("subscribe 1");
                e.onNext(1);
                logD("subscribe 2");
                e.onNext(2);
                logD("subscribe complete");
                e.onComplete();
                logD("subscribe 3");
                e.onNext(3);
            }
        }).subscribe(new io.reactivex.Observer<Integer>() {

            private int i;
            private Disposable mDisposable;

            @Override
            public void onSubscribe(Disposable d) {
                mDisposable = d;
                logD("Observer : onSubscribe: ");
            }

            @Override
            public void onNext(Integer integer) {
                logD("Observer : onNext : " + integer);
                i++;
                if (i == 2) {
                    // 在RxJava 2.x 中，新增的Disposable可以做到切断的操作，让Observer观察者不再接收上游事件
                    mDisposable.dispose();
                }
            }

            @Override
            public void onError(Throwable e) {
                logD("Observer : onError : " + e.toString());
            }

            @Override
            public void onComplete() {
                logD("Observer : complete");
            }
        });
    }

    @OnClick(R.id.btn_schedulers)
    public void onClickSchedulers() {
        //线程切换
        /**
         * subscribeOn() 指定的就是发射事件的线程，observerOn 指定的就是订阅者接收事件的线程。
         * 多次指定发射事件的线程只有第一次指定的有效，也就是说多次调用 subscribeOn() 只有第一次的有效，其余的会被忽略。
         * 但多次指定订阅者接收线程是可以的，也就是说每调用一次 observerOn()，下游的线程就会切换一次。
         *
         * 类型：
         * Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作；
         * Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作；
         * Schedulers.newThread() 代表一个常规的新线程；
         * AndroidSchedulers.mainThread() 代表Android的主线程
         */
        io.reactivex.Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                logD("Schedulers: subscribe: threadName: " + Thread.currentThread().getName() + " priority: " + Thread.currentThread().getPriority());
                logD("subscribe 1");
                e.onNext(1);
                logD("subscribe 2");
                e.onNext(2);
                logD("subscribe complete");
                e.onComplete();
                logD("subscribe 3");
                e.onNext(3);
            }
        }).subscribeOn(Schedulers.io())
                .subscribeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        logD("Schedulers: Consumer: accept: threadName: " + Thread.currentThread().getName() + " priority: " + Thread.currentThread().getPriority());
                        logD("Schedulers: Consumer: accept: " + integer);
                    }
                });
    }

    @OnClick(R.id.btn_consumer)
    public void onClickConsumer() {
        // 单个观察者回调，只执行onNext、onComplete
        io.reactivex.Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                logD("subscribe 1");
                e.onNext(1);
                logD("subscribe 2");
                e.onNext(2);
                logD("subscribe complete");
                e.onComplete();
                logD("subscribe 3");
                e.onNext(3);
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                logD("Consumer : " + integer);
            }

        });
    }

    public void logD(String msg) {
        Log.d(TAG, msg);
    }
}
